package com.millstein.realtime.app.dwd.db;

import com.millstein.realtime.app.base.BaseSqlApp;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Description
 * @Author tsing
 * @Date 2024-10-12 14:31
 */
public class Dwd_10_TradeRefundPaySuc extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_10_TradeRefundPaySuc().init(
                7004,
                2,
                "Dwd_10_TradeRefundPaySuc");
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env      执行环境
     * @param tableEnv 表执行环境
     */
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1.设置数据过期时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        // 2.读取ods层的数据
        readOdsDataFromKafka(tableEnv, "Dwd_10_TradeRefundPaySuc");

        // 3.从ods层数据中筛选出退单数据。因为退单里有退单数量，退款里没有。后续要根据退款商品数量进行分析
        Table orderRefundInfo = tableEnv.sqlQuery(
                "select " +
                "    `data`['order_id'] order_id, " +
                "    `data`['sku_id'] sku_id, " +
                "    `data`['refund_num'] refund_num " +
                "from maxwell_table " +
                "where `database` = 'gmall' " +
                "    and `table` = 'order_refund_info' " +
                "    and `type` = 'update' " +
                "    and `old`['refund_status'] is not null " +
                "    and `data`['refund_status'] = '0705'"
        );
        tableEnv.createTemporaryView("order_refund_info", orderRefundInfo);

        // 4.从ods层数据中筛选出订单数据。订单中有省份id，后续有需求根据省份id进行分析
        Table orderInfo = tableEnv.sqlQuery(
                "select " +
                "    `data`['id'] order_id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['province_id'] province_id " +
                "from maxwell_table " +
                "where `database` = 'gmall' " +
                "    and `table` = 'order_info' " +
                "    and `type` = 'update' " +
                "    and `old`['order_status'] is not null " +
                "    and `data`['order_status'] = '1006'"
        );
        tableEnv.createTemporaryView("order_info", orderInfo);

        // 5.从ods层数据中筛选出退款数据
        Table refundPayment = tableEnv.sqlQuery(
                "select " +
                "    `data`['id'] id, " +
                "    `data`['order_id'] order_id, " +
                "    `data`['sku_id'] sku_id, " +
                "    `data`['payment_type'] payment_type, " +
                "    `data`['total_amount'] total_amount, " +
                "    `data`['callback_time'] callback_time, " +
                "    `pt`, " +
                "    `ts` " +
                "from maxwell_table " +
                "where `database` = 'gmall' " +
                "    and `table` = 'refund_payment' " +
                "    and `type` = 'update' " +
                "    and `old`['refund_status'] is not null " +
                "    and `data`['refund_status'] = '0705'"
        );
        tableEnv.createTemporaryView("refund_payment", refundPayment);

        // 6.从数据库中读取base_dic的数据
        readBaseDicFromMysql(tableEnv);

        // 7.四表join
        Table resultTable = tableEnv.sqlQuery(
                "select " +
                "    rp.id, " +
                "    rp.order_id, " +
                "    oi.user_id, " +
                "    oi.province_id, " +
                "    rp.sku_id, " +
                "    rp.payment_type payment_type_code, " +
                "    bd.dic_name payment_type_name, " +
                "    ori.refund_num, " +
                "    rp.total_amount, " +
                "    date_format(rp.callback_time, 'yyyy-MM-dd') date_id, " +
                "    rp.callback_time, " +
                "    rp.ts, " +
                "    current_row_timestamp() row_opt_ts " +
                "from refund_payment rp " +
                "join order_refund_info ori on rp.order_id = ori.order_id and rp.sku_id = ori.sku_id " +
                "join order_info oi on rp.order_id = oi.order_id " +
                "join base_dic for system_time as of rp.pt as bd on rp.payment_type = bd.dic_name"
        );

        // 8.创建kafka-sink的动态表
        tableEnv.executeSql(
                "create table dwd_refund_pay_suc ( " +
                "    id string, " +
                "    order_id string, " +
                "    user_id string, " +
                "    province_id string, " +
                "    sku_id string, " +
                "    payment_type_code string, " +
                "    payment_type_name string, " +
                "    refund_num string, " +
                "    total_amount string, " +
                "    date_id string, " +
                "    callback_time string, " +
                "    ts bigint, " +
                "    row_opt_ts timestamp_ltz(3) " +
                ") with ( " +
                "    'connector' = 'kafka', " +
                "    'properties.bootstrap.servers' = 'hadoop102:9092,hadoop103:9092,hadoop104:9092', " +
                "    'topic' = 'dwd_refund_pay_suc', " +
                "    'format' = 'json' " +
                ")"
        );

        // 9.将最终数据写入kafka中
        resultTable.executeInsert("dwd_refund_pay_suc");
    }
}
